package com.xxwu.messagequeue.mqclient;

import com.xxwu.messagequeue.common.*;
import com.xxwu.messagequeue.common.argumentsType.*;
import com.xxwu.messagequeue.mqserver.core.BasicProperties;
import com.xxwu.messagequeue.mqserver.core.ExchangeType;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.HashMap;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
@Slf4j
@Data
public class Channel {
    private String channelId;
    // 当前 channel 属于哪个连接,将两个类的关系建立起来
    private Connection connection;
    // 记录当前收到的响应
    private ConcurrentHashMap<String, BasicReturns> returnsMap = new ConcurrentHashMap<>();
    // 记录当前对应的回调函数，一个 channel 只能对应一个回调函数
    private Consumer consumer;

    public Channel(String channelId, Connection connection) {
        this.channelId = channelId;
        this.connection = connection;
    }

    // 通知服务器创建一个 channel
    public Boolean createChannel() throws IOException {
        // 构造请求体
        BasicArguments basicArguments = new BasicArguments();
        basicArguments.setChannelId(channelId);
        basicArguments.setRid(createRid());
        // 将请求体序列化用于传输
        byte[] payload = BinaryTool.tobytes(basicArguments);

        // 创建请求对象
        Request request = new Request();
        request.setType(0x1);
        request.setLength(payload.length);
        request.setPayload(payload);

        // 发送请求
        connection.writeRequest(request);

        // 一个 connection 可能对应着多个 channel，多个 channel 可能都向服务器发送了请求
        // 但服务器响应的顺序不一定与发送顺序相同，这时就要用到最开始创建的 returnsMap
        // 当服务器返回响应后，就将响应先存储在 returnsMap 中
        // 创建一个线程扫描这个 returnsMap，当 returnsMap 中有了与当前请求对应的响应时，这个线程才能继续运行
        // 对应关系通过 rid 实现

        // 阻塞等待响应结果
        BasicReturns basicReturns = waitResponds(basicArguments.getRid());
        return basicReturns.isOk();
    }

    private BasicReturns waitResponds(String rid) {
        // 去 returnsMap 中找有没有 rid 对应的 BasicReturns,若没有找到就要阻塞线程
        BasicReturns basicReturns = null;
        while(returnsMap.get(rid) == null) {
            synchronized (this) {
                try {
                    // 直到有新响应返回，才将它唤醒
                    wait();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }
        basicReturns = returnsMap.get(rid);
        // 获取完之后将 响应从集合中删除
        returnsMap.remove(rid);
        return basicReturns;
    }

    public void putReturns(BasicReturns basicReturns) {
        returnsMap.put(basicReturns.getRid(), basicReturns);
        synchronized (this) {
            // 当前也不知道有多少个线程在等待响应，将所有都唤醒
            // 只有一个线程能拿到结果，其他都继续等待
            notifyAll();
        }
    }
    // 删除 channel
    public Boolean close() throws IOException {
        // 构造请求体
        BasicArguments basicArguments = new BasicArguments();
        basicArguments.setChannelId(channelId);
        basicArguments.setRid(createRid());
        byte[] payload = BinaryTool.tobytes(basicArguments);
        // 构造请求
        Request request = new Request();
        request.setType(0x2);
        request.setLength(payload.length);
        request.setPayload(payload);
        connection.writeRequest(request);
        BasicReturns basicReturns = waitResponds(basicArguments.getRid());
        return basicReturns.isOk();
    }

    private String createRid() {
        return "R-" + UUID.randomUUID();
    }

    public boolean exchangeDeclare(String exchangeName, ExchangeType type,
                                Boolean durable, Boolean autoDelete, HashMap<String, Object> argument) throws IOException {
        // 构造请求体
        ExchangeDeclareArguments exchangeDeclareArguments = new ExchangeDeclareArguments();
        exchangeDeclareArguments.setChannelId(channelId);
        exchangeDeclareArguments.setRid(createRid());
        exchangeDeclareArguments.setExchangeName(exchangeName);
        exchangeDeclareArguments.setType(type);
        exchangeDeclareArguments.setDurable(durable);
        exchangeDeclareArguments.setAutoDelete(autoDelete);
        exchangeDeclareArguments.setArgument(argument);

        // 将请求体进行序列化
        byte[] payload = BinaryTool.tobytes(exchangeDeclareArguments);
        Request request = new Request();
        request.setType(0x3);
        request.setLength(payload.length);
        request.setPayload(payload);

        // 发送请求
        connection.writeRequest(request);

        // 阻塞等待响应
        BasicReturns basicReturns = waitResponds(exchangeDeclareArguments.getRid());
        return basicReturns.isOk();
    }

    public boolean exchangeDelete(String exchangeName) throws IOException {
        ExchangeDeleteArguments exchangeDeleteArguments = new ExchangeDeleteArguments();
        exchangeDeleteArguments.setChannelId(channelId);
        exchangeDeleteArguments.setRid(createRid());
        exchangeDeleteArguments.setExchangeName(exchangeName);

        byte[] payload = BinaryTool.tobytes(exchangeDeleteArguments);

        Request request = new Request();
        request.setType(0x4);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        log.info("ClientRequest.type={},responds.payload={}", request.getType(),request.getPayload());
        BasicReturns basicReturns = waitResponds(exchangeDeleteArguments.getRid());
        return basicReturns.isOk();
    }

    public boolean queueDeclare(String queueName, Boolean durable, Boolean autoDelete
            , Boolean exclusive, HashMap<String, Object> argument) throws IOException {
        QueueDeclareArguments queueDeclareArguments = new QueueDeclareArguments();
        queueDeclareArguments.setChannelId(channelId);
        queueDeclareArguments.setRid(createRid());
        queueDeclareArguments.setQueueName(queueName);
        queueDeclareArguments.setDurable(durable);
        queueDeclareArguments.setAutoDelete(autoDelete);
        queueDeclareArguments.setExclusive(exclusive);
        queueDeclareArguments.setArgument(argument);

        byte[] payload = BinaryTool.tobytes(queueDeclareArguments);

        Request request = new Request();
        request.setType(0x5);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicReturns basicReturns = waitResponds(queueDeclareArguments.getRid());
        return basicReturns.isOk();
    }

    public boolean queueDelete(String queueName) throws IOException {
        QueueDeleteArguments queueDeleteArguments = new QueueDeleteArguments();
        queueDeleteArguments.setChannelId(channelId);
        queueDeleteArguments.setRid(createRid());
        queueDeleteArguments.setQueueName(queueName);

        byte[] payload = BinaryTool.tobytes(queueDeleteArguments);

        Request request = new Request();
        request.setType(0x6);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicReturns basicReturns = waitResponds(queueDeleteArguments.getRid());
        return basicReturns.isOk();
    }

    public boolean queueBind(String exchangeName, String queueName, String bindingKey) throws IOException {
        QueueBindArguments queueBindArguments = new QueueBindArguments();
        queueBindArguments.setChannelId(channelId);
        queueBindArguments.setRid(createRid());
        queueBindArguments.setExchangeName(exchangeName);
        queueBindArguments.setQueueName(queueName);
        queueBindArguments.setBindingKey(bindingKey);

        byte[] payload = BinaryTool.tobytes(queueBindArguments);

        Request request = new Request();
        request.setType(0x7);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicReturns basicReturns = waitResponds(queueBindArguments.getRid());
        return basicReturns.isOk();
    }

    public boolean queueUnbind(String exchangeName, String queueName) throws IOException {
        QueueUnBindArguments queueUnBindArguments = new QueueUnBindArguments();
        queueUnBindArguments.setChannelId(channelId);
        queueUnBindArguments.setRid(createRid());
        queueUnBindArguments.setExchangeName(exchangeName);
        queueUnBindArguments.setQueueName(queueName);

        byte[] payload = BinaryTool.tobytes(queueUnBindArguments);

        Request request = new Request();
        request.setType(0x8);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicReturns basicReturns = waitResponds(queueUnBindArguments.getRid());
        return basicReturns.isOk();
    }

    public boolean basicPublish(String exchangeName, String routingKey
            , BasicProperties basicProperties, byte[] body) throws IOException {
        BasicPublishArguments basicPublishArguments = new BasicPublishArguments();
        basicPublishArguments.setChannelId(channelId);
        basicPublishArguments.setRid(createRid());
        basicPublishArguments.setExchangeName(exchangeName);
        basicPublishArguments.setRoutingKey(routingKey);
        basicPublishArguments.setBasicProperties(basicProperties);
        basicPublishArguments.setBody(body);

        byte[] payload = BinaryTool.tobytes(basicPublishArguments);

        Request request = new Request();
        request.setType(0x9);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicReturns basicReturns = waitResponds(basicPublishArguments.getRid());
        return basicReturns.isOk();
    }

    public boolean basicConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException, IOException {
        BasicConsumerArguments basicConsumerArguments = new BasicConsumerArguments();
        basicConsumerArguments.setChannelId(channelId);
        basicConsumerArguments.setConsumerTag(consumerTag);
        basicConsumerArguments.setRid(createRid());
        basicConsumerArguments.setQueueName(queueName);
        basicConsumerArguments.setAutoAck(autoAck);

        // 这里并不会把 consumer 发送给服务器，而是保存下来，用于后续消费消息
        if(this.consumer != null) {
            throw new MqException("consumer 已经设置过！");
        }
        this.consumer = consumer;

        byte[] payload = BinaryTool.tobytes(basicConsumerArguments);

        Request request = new Request();
        request.setType(0xa);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicReturns basicReturns = waitResponds(basicConsumerArguments.getRid());
        return basicReturns.isOk();
    }

    public boolean basicAck(String queueName, String messageId) throws IOException {
        BasicAckArguments basicAckArguments = new BasicAckArguments();
        basicAckArguments.setChannelId(channelId);
        basicAckArguments.setRid(createRid());
        basicAckArguments.setMessageId(messageId);

        byte[] payload = BinaryTool.tobytes(basicAckArguments);

        Request request = new Request();
        request.setType(0xb);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicReturns basicReturns = waitResponds(basicAckArguments.getRid());
        return basicReturns.isOk();
    }
}
